package org.study.binlog4j.boot.config;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;

/**
 * 作用描述
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2024年12月11日
 */
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
@Component
@EnableConfigurationProperties({BinaryLogProperties.class})
public class BinaryLogClientInit implements InitializingBean, DisposableBean {
    private final BinaryLogProperties binaryLogProperties;
    private BinaryLogClient client;

    /**
     * 根据message获取对象
     */
    private JSONObject getDataObject(List<Serializable> message) {
        JSONObject resultObject = new JSONObject();
        String format = "{\"id\":\"0\",\"name\":\"1\",\"age\":\"2\",\"code\":\"3\"}";
        JSONObject json = JSON.parseObject(format);
        for (String key : json.keySet()) {
            resultObject.put(key, message.get(json.getInteger(key)));
        }
        return resultObject;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("afterPropertiesSet()当前线程名称：{}", Thread.currentThread().getName());
        // 这里的账号必须要有权限访问
        client = new BinaryLogClient(binaryLogProperties.getHost(), binaryLogProperties.getPort(), binaryLogProperties.getUsername(), binaryLogProperties.getPassword());
        // 反序列化配置
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG
                // EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        // 设置反序列化配置
        client.setEventDeserializer(eventDeserializer);
        // 设置自己的client作为服务器的id
        client.setServerId(3);
        // 可选，设置start fileName+position
        // client.setBinlogFilename("master-bin.000080");
        // client.setBinlogPosition(219);
        client.registerEventListener(event -> {
            EventData data = event.getData();
            String tableName;
            if (data instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) data;
                log.info("tableId：{}，database：{}，table：{}", tableMapEventData.getTableId(), tableMapEventData.getDatabase(), tableMapEventData.getTable());
                tableName = tableMapEventData.getTable();
                // 如果是不处理的表，那么返回
                if (!Objects.equals(tableName, "student"))
                    return;
            }
            if (data instanceof UpdateRowsEventData) {
                log.info("Update：{}", data);
                // 获取对应的操作对象的json化数据
                UpdateRowsEventData update = (UpdateRowsEventData) data;
                List<Entry<Serializable[], Serializable[]>> rows = update.getRows();
                for (Map.Entry<Serializable[], Serializable[]> row : rows) {
                    List<Serializable> entries = Arrays.asList(row.getValue());
                    JSONObject dataObject = getDataObject(entries);
                    log.info("Update dataObject：{}", dataObject);
                }
            } else if (data instanceof WriteRowsEventData) {
                log.info("Insert：{}", data);
                WriteRowsEventData wData = new WriteRowsEventData();
                wData.getIncludedColumns();
                wData.getRows();
            } else if (data instanceof DeleteRowsEventData) {
                log.info("Delete：{}", data);
            }
        });
        CompletableFuture.runAsync(() -> {
            try {
                client.connect();
            } catch (Exception ex) {
                log.error("BinaryLogClient 监听出现异常", ex);
                System.exit(-1);
            }
        });
    }

    @Override
    public void destroy() throws Exception {
        log.info("destroy()当前线程名称：{}", Thread.currentThread().getName());
        if (client != null && client.isConnected()) {
            client.disconnect();
            log.info("断开BinaryLogClient连接成功");
        }
    }
}
